package com.atguigu.flink.streamapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * Created by Smexy on 2023/1/30
 *
 *      很少会用到。
 *              测试时会用到
 */
public class Demo4_CustomSink
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .keyBy(WaterSensor::getId)
                   .sum("vc")
                   .addSink(new MySink());

                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    /*
            自定义的Sink必须是 SinkFunction类型。

            写出到外部设备一般都需要获取链接(每个Task只需要执行一次)
                生命周期方法！
                    RichXXX

           模拟向Mysql发送数据。
                存每种传感器累积的vc值

         -----------
CREATE TABLE `ws` (
  `id` varchar(100) NOT NULL,
  `ts` bigint(20) DEFAULT NULL,
  `vc` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8
     */
    private static class MySink extends RichSinkFunction<WaterSensor>{

        private Connection connection;

        String sql = "replace into ws values(?,?,?) ";
        private PreparedStatement ps;

        @Override
        public void invoke(WaterSensor value, Context context) throws Exception {

                ps.setString(1,value.getId());
                ps.setLong(2,value.getTs());
                ps.setInt(3,value.getVc());

                //执行写入
                ps.executeUpdate();
        }

        //获取链接
        @Override
        public void open(Configuration parameters) throws Exception {

            connection = DriverManager.getConnection(
                "jdbc:mysql://hadoop104:3306/220828?useSSL=false&useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true",
                "root",
                "000000"
            );

            ps = connection.prepareStatement(sql);
        }

        //关闭链接
        @Override
        public void close() throws Exception {
            if (connection != null){
                connection.close();
            }

        }
    }
}
